package net.qiujuer.library.clink.impl.async;

import net.qiujuer.library.clink.core.IoArgs;
import net.qiujuer.library.clink.core.SendDispatcher;
import net.qiujuer.library.clink.core.SendPacket;
import net.qiujuer.library.clink.core.Sender;
import net.qiujuer.library.clink.utils.CloseUtils;

import java.io.IOException;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

public class AsyncSendDispatcher implements SendDispatcher {
    private final AtomicBoolean isClosed = new AtomicBoolean(false);
    private final Sender sender;
    private final Queue<SendPacket> queue = new ConcurrentLinkedQueue<>();
    // 发送过程当前有数据来的时候，我们要Packet丢到queue当中去，然后从queue当中把每一个Packet拿出来发送，而如果说当前正在
    // 进行一个发送状态。当咋们没有数据发送的时候，也就是说我们不再去监听发送的空闲状态，这个时候我们的发送状态是一个关闭状态
    // 知道有新数据来的时候，我们会重新去激活这个状态，所以我们需要一个变量来维护这个状态。
    private final AtomicBoolean isSending = new AtomicBoolean();

    private IoArgs ioArgs = new IoArgs();
    private SendPacket packetTemp;

    // 因为Packet有可能会比IoArgs更大，所以需要一个进度来维护他，这里定义一个当前Packet最大的值。
    private int total;
    // 当前Packet发送了多长
    private int position;

    public AsyncSendDispatcher(Sender sender) {
        this.sender = sender;
    }

    @Override
    public void send(SendPacket packet) {
        queue.offer(packet);
        // 如果当前不在发送中，就设置为发送k中
        if (isSending.compareAndSet(false,true)){
            sendNextPacket();
        }
    }

    private SendPacket takePacket() {
        SendPacket packet = queue.poll();
        if (packet != null && packet.isCanceled()) {
            // 已取消，不用发送，递归拿下一条
            return takePacket();
        }
        return packet;
    }

    // 发送一条数据
    private void sendNextPacket() {
        SendPacket temp = packetTemp;
        // 我们要发送下一条的时候，发现我们当前的发送的这条还不等于空
        if (temp!=null) {
            CloseUtils.close(temp);
        }

        SendPacket packet = takePacket();
        if (packet == null) {
            // 队列为空，取消状态发送
            isSending.set(false);
            return;
        }

        total = packet.length();
        position = 0;
        sendCurrentPacket();
    }

    private void sendCurrentPacket() {
        IoArgs args = ioArgs;

        // 开始，清理
        args.startWriting();
        // 代表数据已经完全发送完毕
        if (position>=total) {
            sendNextPacket();
            return;
        }else if (position == 0) {
            // 首包，需要携带长度信息
            args.writeLength(total);
        }

        byte[] bytes = packetTemp.bytes();
        // 把bytes的数据写入到IoArgs里面
        int count = args.readFrom(bytes, position);
        position += count;

        // 完成封装
        args.finishWriting();

        try {
            sender.sendASync(args, ioArgsEventListener);
        } catch (IOException e) {
            closeAndNotify();
        }

    }

    private void closeAndNotify() {
        CloseUtils.close(this);
    }

    @Override
    public void cancel(SendPacket packet) {

    }

    @Override
    public void close() throws IOException {
        if (isClosed.compareAndSet(false, true)) {
            isSending.set(false);
            SendPacket packet = this.packetTemp;
            if (packet != null) {
                packetTemp = null;
                CloseUtils.close(packet);
            }
        }
    }

    private final IoArgs.IoArgsEventListener ioArgsEventListener = new IoArgs.IoArgsEventListener() {
        @Override
        public void onStarted(IoArgs args) {

        }

        @Override
        public void onCompleted(IoArgs args) {
            // 继续发送当前包
            sendCurrentPacket();
        }
    };
}
